/*
 * Copyright 2008 Wink Saville
 * 
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License version 3
 * as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

package com.saville.mc;

import java.lang.Thread;
import java.net.*;
import java.io.*;
import java.util.Locale;
import java.util.HashMap;
import java.util.concurrent.atomic.*;

import com.saville.debug.*;
import com.saville.msgcomp.*;
import com.saville.serdes.*;

public class HttpMsgPipe {

    /*
     * Constructor
     * 
     * Open a MsgPipe using an existing socket
     */
    public HttpMsgPipe(Socket clientSocket, int producerDstId, int producerDstSubId) {
        mHttpMsgPipeName = "HttpMsgPipe-" + Timing.readTsc();
        Log.d(df, "%s: HttpMsgPipe(clientSocket, producerMc) E", mHttpMsgPipeName);
        init(clientSocket, producerDstId, producerDstSubId);
        Log.d(df, "%s: HttpMsgPipe(clientSocket, producerMc) X", mHttpMsgPipeName);
    }

    /*
     * Constructor
     * 
     * Open a MsgPipe using the url
     */
    public HttpMsgPipe(String clientUrl, int procuderDstId, int producerDstSubId) {
        if (!open(clientUrl, procuderDstId, producerDstSubId)) {
            throw new RuntimeException("HttpMsgPipe: Unable to open clientUrl");
        }
    }

    /*
     * Queue a message which will be sent down the pipe.
     */
    public void queueMsg(Msg msg) {
        mConsumer.queueMsg(msg);
    }

    /*
     * Open
     * 
     * TODO: Maybe make async and return a message when complete?
     */
    private boolean open(String clientUrl, int producerDstId, int producerDstSubId) {
        Socket clientSocket = null;
        boolean result;

        mDstId = producerDstId;
        mDstSubId = producerDstSubId;

        /* TODO: Maybe throw error rather than returning boolean? */
        try {
            URI uri = new URI(clientUrl);
            String scheme = uri.getScheme();
            int port = uri.getPort();
            if ((scheme != null) && (scheme.compareToIgnoreCase("amc") == 0)) {
                clientSocket = new Socket();
                if (port == -1) {
                    port = 46000;
                }
                Log.print("%s: connect to host=%s port=%d", mHttpMsgPipeName, uri.getHost(), port);
                clientSocket.connect(new InetSocketAddress(uri.getHost(), port));
                result = init(clientSocket, producerDstId, producerDstSubId);
            } else {
                result = false;
            }
        } catch (URISyntaxException uriE) {
            Log.print("%s: RISyntaxException e=%s", mHttpMsgPipeName, uriE);
            result = false;
        } catch (IOException e) {
            Log.e(df, "Error opening socket");
            result = false;
        }

        if (!result) {
            sendClosed();
        }

        return result;
    }

    /*
     * Initialize
     * 
     * TODO: Maybe make async and return a message when complete?
     */
    private boolean init(Socket clientSocket, int dstId, int dstSubId) {
        Log.d(df, "%s: init(clientSocket, producerMc) E", mHttpMsgPipeName);
        mMcMgr = McMgr.getInstance();

        mSocket = clientSocket;
        mDstId = dstId;
        mDstSubId = dstSubId;

        mProducer = new HttpPipeProducer();
        mConsumer = new HttpPipeConsumer();

        Log.d(df, "%s: init(clientSocket, producerMc) X", mHttpMsgPipeName);
        return true;
    }

    /*
     * Send one and only one CMD_CLOSED
     */
    private void sendClosed() {
        if (mOpened.getAndSet(false)) {
            Msg pipeMsg = new Msg();
            pipeMsg.guid = McConst.MCPIPE_GUID;
            pipeMsg.cmd = McConst.MCPIPE_CMD_CLOSED;
            pipeMsg.dstId = mDstId;
            pipeMsg.dstSubId = mDstSubId;
            mMcMgr.sendMsg(pipeMsg);
        }
    }

    /*
     * This class reads the pipe of messages sent by the corresponding
     * Consumer and produces messages which are sent to the mDstId with
     * mDstSubId.
     */
    private class HttpPipeProducer implements Runnable {
        public HttpPipeProducer() {
            mHttpPipeProducerName = "HttpPipeProducer" + mSocket.getLocalPort();
            Log.v(df, "%s: HttpPipeProducer(); E", mHttpPipeProducerName);

            Thread thr = new Thread(null, this, mHttpPipeProducerName);
            thr.start();

            Log.v(df, "%s: HttpPipeProducer(); X", mHttpPipeProducerName);
        }
        
        void addHeader(HashMap<String, String> headers, String key, String value) {
            if (headers.containsKey(key)) {
                StringBuffer valueBuffer = new StringBuffer(headers.get(key));
                valueBuffer.append(",");
                valueBuffer.append(value);
                headers.put(key, valueBuffer.toString());
                Log.print("more header key='%s' value='%s'", key, valueBuffer.toString());
            } else {
                Log.print("new header key='%s' value='%s'", key, value);
                headers.put(key, value);
            }
        }
        
        void addHeader(HashMap<String, String> headers, String line) {
            String kv_pair[] = line.split(":", 2);
            if (kv_pair.length != 2) {
                Log.print("bad header no ':' line='%s'", line);
            } else {
                String key = kv_pair[0].trim().toLowerCase(Locale.ENGLISH);
                String value = kv_pair[1].trim();
                addHeader(headers, key, value);
            }
        }
        
        byte nextByte() throws IOException {
            byte b = 0;
            
            if (mPushedByteValid) {
                b = mPushedByte;
                mPushedByteValid = false;
            } else {
                b = mInStream.readByte();
            }
            return b;
        }
        
        void pushByte(byte b) {
            mPushedByteValid = true;
            mPushedByte = b;
        }
        byte mPushedByte;
        boolean mPushedByteValid = false;
        
        String readLine(DataInputStream strm) throws IOException {
            byte b;
            byte nb;
            StringBuffer sb = new StringBuffer();
            
            while(true) {
                b = nextByte();
                
                if (b == '\r') {
                    nb = nextByte();
                    if (nb != '\n')
                        pushByte(nb);
                    break;
                } else if (b == '\n') {
                    nb = nextByte();
                    if (nb != '\r') {
                        pushByte(nb);
                    }
                    break;
                }
                sb.append((char)b);
            }
            
            return sb.toString();
        }
     
        public void run() {
            byte data[] = new byte[16384];
            /*
             * We could handle either json or binary if we either
             * encode either when we open the stream or using a
             * header or even with each message?
             */
            SerdesJson serdes = new SerdesJson();
            int MAX_LENGTH = 65536;

            try {
                Log.v(df, "%s: HttpPipeProducer; run() E", mHttpPipeProducerName);

                SocketAddress sa = mSocket.getRemoteSocketAddress();
                InetSocketAddress isa = (InetSocketAddress) sa;
                Log.v(df, "%s: remote inetAddress=%s, port=%s", mHttpPipeProducerName, isa
                        .getAddress().getHostAddress(), isa.getPort());

                mInStream = new DataInputStream(mSocket.getInputStream());

                while (true) {
                    HashMap<String, String> headers = new HashMap<String, String>();
                    StringBuffer prevStringBuffer = null;
                    String prevLine = null;
                    String line;
                    
                    /*
                     * Read request
                     */
                    String request = readLine(mInStream);
                    Log.print("request='%s', request", request);
                    if (request == null)
                        break;
                    String request_tokens[] = request.split(" ");
                    if ((request_tokens.length == 2) || (request_tokens.length == 3)) {
                        addHeader(headers, "REQ_METHOD", request_tokens[0].trim());
                        addHeader(headers, "REQ_URI", request_tokens[1].trim());
                        if (request_tokens.length == 3) {
                            addHeader(headers, "REQ_VER", request_tokens[2]);
                        } else {
                            addHeader(headers, "REQ_VER", "HTTP/1.0");
                        }
                    } else {
                        throw new RuntimeException("Bad request='" + request + "'");
                    }
                    
                    /*
                     * Read headers
                     */
                    while (((line = readLine(mInStream)) != null) && (line.length() > 0)) {
                        Log.print("read line='%s'", line);
                        char firstCh = line.charAt(0);
                        if ((firstCh == ' ') || (firstCh == '\t')) {
                            if (prevLine != null) {
                                if (prevStringBuffer == null) {
                                    prevStringBuffer = new StringBuffer(prevLine);
                                }
                                prevStringBuffer.append(line);
                                Log.print("continuation line='%s'", prevStringBuffer.toString());
                            } else {
                                Log.print("no previous line, illgal continuation line='%s'", line);
                            }
                        } else {
                            if (prevLine != null) {
                                if (prevStringBuffer != null) {
                                    prevLine = prevStringBuffer.toString();
                                }
                                addHeader(headers, prevLine);
                            }
                            prevLine = line;
                        }
                    }
                    Log.print("done prevLine='%s'", prevLine);
                    if (prevLine != null) {
                        addHeader(headers, prevLine);
                        prevLine = null;
                    }
                    Log.print("headers=%s", headers.toString());
                    
                    /*
                     * Read body
                     */
                    String te = headers.get("Transfer-Encoding");
                    if (te != null) {
                        throw new RuntimeException("Transfer-Encoding not support te='" + te + "'");
                    } else {
                        String cls = headers.get("content-length");
                        if (cls != null) {
                            int len = Integer.parseInt(cls);
                            if (len > MAX_LENGTH) {
                                throw new Exception("Length too large");
                            }
                            if (len > data.length) {
                                data = new byte[len];
                            }
                            mInStream.readFully(data, 0, len);
                            serdes.init(data, 0, len);
                            Msg msg = (Msg) serdes.readObject();
                            Log.v(df, "%s: HttpPipeProducer; got msg guid=0x%x cmd=0x%x", mHttpPipeProducerName,
                                msg.guid, msg.cmd);
                            
                            Msg pipeMsg = new Msg();
                            pipeMsg.guid = McConst.MCPIPE_GUID;
                            pipeMsg.cmd = McConst.MCPIPE_CMD_PRODUCER_HTTP_HEADER_MSG;
                            pipeMsg.dstId = mDstId;
                            pipeMsg.dstSubId = mDstSubId;
                            pipeMsg.obj = headers;
                            mMcMgr.sendMsg(pipeMsg);
                            
                            pipeMsg = new Msg();
                            pipeMsg.guid = McConst.MCPIPE_GUID;
                            pipeMsg.cmd = McConst.MCPIPE_CMD_PRODUCER_MSG;
                            pipeMsg.dstId = mDstId;
                            pipeMsg.dstSubId = mDstSubId;
                            pipeMsg.obj = msg;
                            mMcMgr.sendMsg(pipeMsg);
                        } else {
                            Msg pipeMsg = new Msg();
                            pipeMsg.guid = McConst.MCPIPE_GUID;
                            pipeMsg.cmd = McConst.MCPIPE_CMD_PRODUCER_HTTP_HEADER_MSG;
                            pipeMsg.dstId = mDstId;
                            pipeMsg.dstSubId = mDstSubId;
                            pipeMsg.obj = headers;
        
                            mMcMgr.sendMsg(pipeMsg);
                        }
                    }
                }
            } catch (Exception e) {
                Log.e(df, "%s: HttpPipeProducer; Exception e=%s", mHttpPipeProducerName, e);
                sendClosed();
            } finally {
                try {
                    if (mInStream != null)
                        mInStream.close();
                } catch (Exception e) {
                    Log.e(df, "%s: HttpPipeProducer; exception closing e=%s", mHttpPipeProducerName, e);
                }

                Log.v(df, "%s: HttpPipeProducer; run() X", mHttpPipeProducerName);
            }
        }

        private DataInputStream mInStream;
        private String mHttpPipeProducerName;
    }

    /*
     * This class consumes messages on its queue forwarding them to the producer
     * on the other side.
     */
    private class HttpPipeConsumer extends ActiveMc {
        public HttpPipeConsumer() {
            super("HttpPipeConsumer" + mSocket.getLocalPort());
            df = Log.VERBOSE;

            Log.v(df, "%s: HttpPipeConsumer(); E", mName);

            mSerdes = new SerdesJson();

            try {
                OutputStream os = mSocket.getOutputStream();
                mOutStream = new DataOutputStream(os);
                // mOutStream = new DataOutputStream(mSocket.getOutputStream());
            } catch (Exception e) {
                Log.e(df, "%s: HttpPipeConsumer; Exception e=%s", mName, e);
                sendClosed();
            }

            Log.v(df, "%s: HttpPipeConsumer(); X", mName);
        }

        public void processMsg(Msg msg) {

            // Log.v(df, "%s: HttpPipeConsumer processMsg() E", mName);

            if (msg.is(McConst.MC_GUID, McConst.MC_CMD_STARTED)) {
                Log.v(df, "%s: HttpPipeConsumer; got STARTED", mName);
            } else {

                Log.v(df, "%s: HttpPipeConsumer; got msg guid=0x%x cmd=0x%x", mName, msg.guid, msg.cmd);

                try {
                    mSerdes.init();
                    msg.serialize(mSerdes);
                    byte data[] = mSerdes.data();
                    mOutStream.writeBytes(Integer.toString(data.length));
                    mOutStream.write(data, 0, data.length);

                } catch (Exception e) {
                    Log.e(df, "%s: HttpPipeConsumer exception processing e=%s", mName, e);
                    sendClosed();

                    try {
                        if (mOutStream != null)
                            mOutStream.close();
                    } catch (Exception ec) {
                        Log.e(df, "%s: HttpPipeConsumer exception closing ec=%s", mName, ec);
                    }
                }
            }

            // Log.v(df, "%s: HttpPipeConsumer processMsg() X", mName);
        }

        private SerdesJson mSerdes;
        private DataOutputStream mOutStream;
    }

    private AtomicBoolean mOpened = new AtomicBoolean(true);
    private int mDstId;
    private int mDstSubId;
    private Socket mSocket;
    private String mHttpMsgPipeName;
    private HttpPipeProducer mProducer;
    private HttpPipeConsumer mConsumer;
    private int df = Log.VERBOSE;
    private IMcMgr mMcMgr;
}
